package com.Rezar.dbSub.base;

import java.util.ArrayList;
import java.util.List;

import akka.actor.Address;
import akka.actor.typed.ActorRef;
import akka.actor.typed.Behavior;
import akka.actor.typed.javadsl.AbstractBehavior;
import akka.actor.typed.javadsl.ActorContext;
import akka.actor.typed.javadsl.Behaviors;
import akka.actor.typed.javadsl.Receive;
import akka.cluster.ClusterEvent;
import akka.cluster.typed.Cluster;
import akka.cluster.typed.Down;
import akka.cluster.typed.Subscribe;
import lombok.extern.slf4j.Slf4j;

/**
 * 
 * @say little Boy, don't be sad.
 * @name Rezar
 * @time Jul 22, 2020 11:28:18 AM
 * @Desc 些年若许,不负芳华.
 *
 */
@Slf4j
public final class ClusterListener extends AbstractBehavior<ClusterListener.Event> {

	private Cluster cluster;
	private boolean isLeader;
	private Address address;

	private List<Address> unreachableNodeAddress = new ArrayList<>();

	interface Event {
	}

	private static final class ReachabilityChange implements Event {
		final ClusterEvent.ReachabilityEvent reachabilityEvent;

		ReachabilityChange(ClusterEvent.ReachabilityEvent reachabilityEvent) {
			this.reachabilityEvent = reachabilityEvent;
		}
	}

	private static final class LeaderChange implements Event {
		final ClusterEvent.LeaderChanged leaderChangeEvent;

		LeaderChange(ClusterEvent.LeaderChanged leaderChangeEvent) {
			this.leaderChangeEvent = leaderChangeEvent;
		}
	}

	private static final class MemberChange implements Event {
		final ClusterEvent.MemberEvent memberEvent;

		MemberChange(ClusterEvent.MemberEvent memberEvent) {
			this.memberEvent = memberEvent;
		}
	}

	public static Behavior<Event> create() {
		return Behaviors.setup(context -> new ClusterListener(context));
	}

	private ClusterListener(ActorContext<Event> context) {
		super(context);
		Cluster cluster = Cluster.get(context.getSystem());
		this.cluster = cluster;
		this.address = this.cluster.selfMember().address();
		ActorRef<ClusterEvent.MemberEvent> memberEventAdapter = context.messageAdapter(ClusterEvent.MemberEvent.class,
				MemberChange::new);
		cluster.subscriptions().tell(Subscribe.create(memberEventAdapter, ClusterEvent.MemberEvent.class));

		ActorRef<ClusterEvent.ReachabilityEvent> reachabilityAdapter = context
				.messageAdapter(ClusterEvent.ReachabilityEvent.class, ReachabilityChange::new);
		cluster.subscriptions().tell(Subscribe.create(reachabilityAdapter, ClusterEvent.ReachabilityEvent.class));

		ActorRef<ClusterEvent.LeaderChanged> leaderChangeAdapter = context
				.messageAdapter(ClusterEvent.LeaderChanged.class, LeaderChange::new);
		cluster.subscriptions().tell(Subscribe.create(leaderChangeAdapter, ClusterEvent.LeaderChanged.class));

	}

	@Override
	public Receive<Event> createReceive() {
		return newReceiveBuilder().onMessage(ReachabilityChange.class, this::onReachabilityChange)
				.onMessage(MemberChange.class, this::onMemberChange).onMessage(LeaderChange.class, this::onLeaderChange)
				.build();
	}

	private Behavior<Event> onLeaderChange(LeaderChange event) {
		log.info("onLeaderChange: {}", event.leaderChangeEvent.getLeader());
		if (event.leaderChangeEvent.getLeader() != null) {
			this.isLeader = event.leaderChangeEvent.getLeader().equals(this.address);
			log.info("self:{} is leader?{}", this.address, this.isLeader);
			if (this.isLeader && !this.unreachableNodeAddress.isEmpty()) {
				for (int index = 0; index < this.unreachableNodeAddress.size(); index++) {
					this.markDown(this.unreachableNodeAddress.remove(index));
				}
			}
		}
		return this;
	}

	private void markDown(Address address) {
		if (this.isLeader) {
			log.info("leader markDown: {}", address);
			this.cluster.manager().tell(new Down(address));
		}
	}

	private Behavior<Event> onReachabilityChange(ReachabilityChange event) {
		if (event.reachabilityEvent instanceof ClusterEvent.UnreachableMember) {
			log.info("Member detected as unreachable: {}", event.reachabilityEvent.member());
			if (this.isLeader) {
				this.markDown(event.reachabilityEvent.member().address());
			} else {
				unreachableNodeAddress.add(event.reachabilityEvent.member().address());
			}
		} else if (event.reachabilityEvent instanceof ClusterEvent.ReachableMember) {
			log.info("Member back to reachable: {}", event.reachabilityEvent.member());
			if (this.unreachableNodeAddress.contains(event.reachabilityEvent.member().address())) {
				log.info("mark address as Reachable:{}", event.reachabilityEvent.member().address());
				this.unreachableNodeAddress.remove(event.reachabilityEvent.member().address());
			}
		}
		return this;
	}

	private Behavior<Event> onMemberChange(MemberChange event) {
		if (event.memberEvent instanceof ClusterEvent.MemberUp) {
			log.info("Member is up: {}", event.memberEvent.member());
		} else if (event.memberEvent instanceof ClusterEvent.MemberRemoved) {
			log.info("Member is removed: {} after {}", event.memberEvent.member(),
					((ClusterEvent.MemberRemoved) event.memberEvent).previousStatus());
			if (this.unreachableNodeAddress.contains(event.memberEvent.member().address())) {
				log.info("mark address as removed:{}", event.memberEvent.member().address());
				this.unreachableNodeAddress.remove(event.memberEvent.member().address());
			}
		}
		return this;
	}

}
